/*
 * Copyright (C) 2017 Pingan, Inc. All Rights Reserved.
 */

package com.pingan.haofang.searchcloud.es.indices.impl;

import com.pingan.haofang.searchcloud.api.DiskConditonDTO;
import com.pingan.haofang.searchcloud.api.matedata.CollectionMetaData;
import com.pingan.haofang.searchcloud.es.api.client.EsClientFactory;
import com.pingan.haofang.searchcloud.es.indices.IndicesManagerService;
import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsRequest;
import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.shard.DocsStats;
import org.elasticsearch.index.store.StoreStats;
import org.elasticsearch.indices.NodeIndicesStats;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * 索引管理
 *
 * @author zhangbi617
 * @date 2017-07-04
 */
@Service
public class IndicesManagerServiceImpl implements IndicesManagerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(IndicesManagerServiceImpl.class);

    @Autowired
    private EsClientFactory esClientFactory;

    @Override
    public boolean create(String host, CollectionMetaData indicesMetaData) {
        boolean result =
                esClientFactory.getByHost(host).admin().indices()
                        .prepareCreate(indicesMetaData.getName()).setSettings(Settings.builder()
                        .put("index.number_of_shards", indicesMetaData.getShardCount())
                        .put("index.number_of_replicas", indicesMetaData.getReplicationFactor())
                ).get().isAcknowledged();
        return result;
    }

    @Override
    public boolean delete(String host, String indicesName) {
        return esClientFactory.getByHost(host).admin()
                .indices().prepareDelete(indicesName).get().isAcknowledged();


    }

    @Override
    public List<String> getIndicesNames(String host) {
        String[] indices = esClientFactory.getByHost(host).admin().indices().prepareGetIndex().get().getIndices();
        return Arrays.stream(indices).collect(Collectors.toList());
    }

    /**
     * 存活的transport client节点
     *
     * @param clusterHost 集群host
     * @return ip:port
     */
    @Override
    public List<String> getLiveHostUrl(String clusterHost) {
        NodesInfoResponse nodesInfoResponse = esClientFactory.getByHost(clusterHost).admin().cluster().nodesInfo(new NodesInfoRequest().clear().transport(true)).actionGet();
        List<String> collect = Optional.ofNullable(nodesInfoResponse).map(NodesInfoResponse::getNodes).map(list ->
                list.stream().map(NodeInfo::getNode).map(DiscoveryNode::getAddress).map(addr -> addr.toString()).collect(Collectors.toList())
        ).orElse(new ArrayList<>());
        return collect;
    }

    /**
     * @param clusterHost
     * @return
     */
    @Override
    public Map<String, DiskConditonDTO> getNodesSummary(String clusterHost) {
        NodesStatsRequest nodesStatsRequest = new NodesStatsRequest().indices(true);
        NodesStatsResponse nodesStatsResponse = esClientFactory.getByHost(clusterHost).admin().cluster().nodesStats(nodesStatsRequest).actionGet();
        List<NodeStats> nodes = nodesStatsResponse.getNodes();

        Map<String, DiskConditonDTO> param = new HashMap<>();
        for (NodeStats stats : nodes) {
            DiskConditonDTO dto = new DiskConditonDTO();

            NodeIndicesStats indicesStats = stats.getIndices();
            StoreStats store = indicesStats.getStore();
            DocsStats docs = indicesStats.getDocs();
            DiscoveryNode node = stats.getNode();

            dto.setNodeIp(node.getAddress().toString());
            dto.setDiskUsed(BigDecimal.valueOf(store.getSizeInBytes()).divide(BigDecimal.valueOf(1024*1024), 0, BigDecimal.ROUND_HALF_UP).longValue());
            param.put(node.getAddress().toString(), dto);
        }
        return param;
    }


}
